package com.example.data_receiver.service;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.errors.TopicExistsException;
import java.util.Collections;
import java.util.concurrent.ExecutionException;

import java.util.List;

@Service
@RequiredArgsConstructor
@Slf4j
public class KafkaService {

    private final KafkaTemplate<String, Object> kafkaTemplate;

    private final AdminClient adminClient;

    public void sendBatch(String topic, List<Object> data) {
        log.info("Sending batch of {} records to topic {}", data.size(), topic);
        // 判断topic是否存在，不存在则创建
        short replicationFactor=3;
        createTopic(topic,3,replicationFactor);
        data.forEach(record -> {

            kafkaTemplate.send(topic, record);
            log.debug("Record sent to topic {}: {}", topic, record);
        });
        log.info("Batch sent successfully to topic {}", topic);
    }

    /**
     * 创建Kafka Topic并指定副本数
     * @param topicName 主题名称
     * @param numPartitions 分区数
     * @param replicationFactor 副本数
     */
    public void createTopic(String topicName, int numPartitions, short replicationFactor) {
        try {
            // 检查Topic是否已存在
            if (!adminClient.listTopics().names().get().contains(topicName)) {
                NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
                adminClient.createTopics(Collections.singleton(newTopic)).all().get();
                log.info("Topic {} created successfully with {} partitions and {} replicas", 
                         topicName, numPartitions, replicationFactor);
            } else {
                log.info("Topic {} already exists", topicName);
            }
        } catch (InterruptedException | ExecutionException e) {
            if (e.getCause() instanceof TopicExistsException) {
                log.info("Topic {} already exists", topicName);
            } else {
                log.error("Failed to create topic {}", topicName, e);
                throw new RuntimeException("Error creating Kafka topic", e);
            }
        }
    }
}